繼之前講到的OTel Client archetecture中, SDK會有多組件.
Processor是很重要的其中一個.
在Trace這pipeline中, processor組件負責對span做處理,
一起來看看吧!

Span Processor

Span Processor其實是一組interace程式碼連結
提供了一組hook, 讓span start和end階段能調用.
並且只有在IsRecording是true的時候才會去調用Span processors.

OTel內建有一些processor, 負責把span做批次處理或做匯出格式的轉換.
Processors可以在Trace provider上註冊, 如果註冊了多個processor, 則被調用處理的順序跟註冊的順序會一樣.

以下是SpanProcessor的介面,
也是能自定義processor, 只要實做了該介面

type SpanProcessor interface {
	OnStart(parent context.Context, s ReadWriteSpan).
	OnEnd(s ReadOnlySpan)
	Shutdown(ctx context.Context) error
	ForceFlush(ctx context.Context) error
}

上圖是SpanProcessor與SDK的其他組件間的關係

配置多個SpanProcessor於trace provier的範例.
我先配置BatchSpanProcessor, 而後配置SimpleSpanProcessor
Span被處理的順序也是配置順序的

	exp, err := exporterToJaeger()
	if err != nil {
		log.Fatalf("error: %s", err.Error())
	}

	tp := trace.NewTracerProvider(
		trace.WithSampler(getSampler()),
		trace.WithSpanProcessor(trace.NewBatchSpanProcessor(exp)),
		trace.WithSpanProcessor(trace.NewSimpleSpanProcessor(exp)),
		trace.WithResource(newResource(ctx)),
	)
	otel.SetTracerProvider(tp)

依序把配置的processor給按順序配置的程式碼如下
第二段是trace provider的, 會range依序把span processor給註冊進去
spanProcessors於trace provider,則是用GO 1.4有的atomic.value做存儲.

// WithSpanProcessor registers the SpanProcessor with a TracerProvider.
func WithSpanProcessor(sp SpanProcessor) TracerProviderOption {
	return traceProviderOptionFunc(func(cfg tracerProviderConfig) tracerProviderConfig {
		cfg.processors = append(cfg.processors, sp)
		return cfg
	})
}

type TracerProvider struct {
	mu             sync.Mutex
	namedTracer    map[instrumentation.Library]*tracer
	spanProcessors atomic.Value

	// These fields are not protected by the lock mu. They are assumed to be
	// immutable after creation of the TracerProvider.
	sampler     Sampler
	idGenerator IDGenerator
	spanLimits  SpanLimits
	resource    *resource.Resource
}

func NewTracerProvider(opts ...TracerProviderOption) *TracerProvider {
	//...省略
	for _, sp := range o.processors {
		tp.RegisterSpanProcessor(sp)
	}

	return tp
}

OTel Go的sdk裡面也有簡單的方法能啟用SimpleSpanProcessor和BatchSpanProcessor
只要調用下面兩個方法就好

func WithSyncer(e SpanExporter) TracerProviderOption {
	return WithSpanProcessor(NewSimpleSpanProcessor(e))
}

// WithBatcher registers the exporter with the TracerProvider using a
// BatchSpanProcessor configured with the passed opts.
func WithBatcher(e SpanExporter, opts ...BatchSpanProcessorOption) TracerProviderOption {
	return WithSpanProcessor(NewBatchSpanProcessor(e, opts...))
}

Batch Span Processor

這是內建的一個span processor,
有些參數來看看

// Defaults for BatchSpanProcessorOptions.
const (
	DefaultMaxQueueSize       = 2048
	DefaultScheduleDelay      = 5000
	DefaultExportTimeout      = 30000
	DefaultMaxExportBatchSize = 512
)

type BatchSpanProcessorOptions struct {
	MaxQueueSize int
	BatchTimeout time.Duration
	ExportTimeout time.Duration
	MaxExportBatchSize int
	BlockOnQueueFull bool
}

type batchSpanProcessor struct {
	e SpanExporter
	o BatchSpanProcessorOptions

	queue   chan ReadOnlySpan
	dropped uint32

	batch      []ReadOnlySpan
	batchMutex sync.Mutex
	timer      *time.Timer
	stopWait   sync.WaitGroup
	stopOnce   sync.Once
	stopCh     chan struct{}
}

Batch Span Processor有幾個重要參數

• e SpanExporter : exporter for span
• MaxQueueSize : batchSpanProcessor內有個queue成員, queue有可能會滿full, 但你也不能讓這full而導致原本的業務程序停著同步在等這裡消費, 所以設置這數字的行為是, 當queue的span數量達到這數字時, 之後的都會被drop; 預設是2048
• ScheduledDelayMillis : 連續兩次對exporter中間的延遲間隔時長; 單位是millisecond, 預設是5000
• ExportTimeoutMillis : exporter在程序被取消前, 如果還有東西在匯出, 可能透過這參數, 配置來等它完成工作; 單位是millisecond, 預設是30000
• MaxExportBatchSize : 一次批量的最大處理的span數量, 但這配置一定要<= MaxQueueSize; 預設是512

今日小心得

之所以之前沒那別把這些用程式做說明,
是因為文件跟參考在trace這環節比較完整, 因為OpenTracing那邊真的很多文件與說明.
透過trace來把東西講完整點
這裡很多名詞, 組件間的功能與關係, 在OTel collector的配置也會出現.

參考資料

OTel specification - Trace SDK